package com.lzy.gmall.realtime.app.dws;

import com.lzy.gmall.realtime.bean.T2;
import com.lzy.gmall.realtime.bean.T3;
import com.lzy.gmall.realtime.utils.MyClickHouseUtil;
import com.lzy.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
/*
*
*                               指标三 支付金额
*
* */
public class Dws_The_amount_to_be_paid {
    public static void main(String[] args) throws Exception {
//        TODO 1.环境初始化
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        TODO 2.设置并行度
        env.setParallelism(1);
//        TODO 3,设置表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//      Kafka 连接配置
        String topic = "dwd_order_pre";
        String groupId = "dws_payment_buyer_count1";

        // 2. 创建Kafka源表
        tableEnv.executeSql("CREATE TABLE order_source (" +
                "  id STRING," +//订单明细 ID
                "  order_id STRING," +//订单 ID
                "  sku_id STRING," +//商品 SKU ID
                "  sku_name STRING," +//商品名称
                "  order_price double," +//订单价格
                "  sku_num INT," +//商品数量
                "  create_time STRING," +//订单创建时间
                "  source_type_name STRING," +//订单来源名称
                "  total_amount double," +//订单总金额
                "  order_status STRING," +//订单状态
                "  feight_fee double," +//运费
                "  row_op_ts STRING," +
//              将 row_op_ts通过REPLACE转换为 Flink适用戳类型。
                "  `time_ltz` AS TO_TIMESTAMP_LTZ(" +
                "    UNIX_TIMESTAMP(REPLACE(row_op_ts, ' ', 'T'), 'yyyy-MM-dd''T''HH:mm:ss') * 1000," +
                "    3" +
                "  )," +
                "  WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND" +//设置水位线5秒延迟数据
                ")" + MyKafkaUtil.getKafkaDDL(topic, groupId));

//        通过子查询筛选出相关的订单状态
        Table t0 = tableEnv.sqlQuery("with  t1  as( " +
                "select   *   from order_source where order_status='1002' or order_status='1003' or order_status ='1004' or order_status='1005'  " +
                ") SELECT " +
                "  DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') AS stt," +
                "  DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') AS edt, " +
                "  COUNT(DISTINCT order_id) AS order_count, " +
                "  SUM(order_price * sku_num + feight_fee) AS total_payment_amount " +
                "FROM TABLE(" +
//                      滚动窗口每 10 秒生成一个不重叠的窗口统计窗口内的数据
                "  TUMBLE(TABLE t1, DESCRIPTOR(`time_ltz`), INTERVAL '10' SECOND)" +
                ") " +
                "  " +
                "GROUP BY window_start, window_end");
        tableEnv.toAppendStream(t0, T3.class).addSink(MyClickHouseUtil.getSinkFunction("insert into gmall.Dws_The_amount_to_be_paid3 values(?,?,?,?)"));

        env.execute();
    }
}
